/*
 * LoadBalancing.cpp
 *
 *  Created on: 31/01/2014
 *      Author: root
 */

#include "LoadBalancing.h"
#include "Utils.h"
//Sacar a archivo de configuraciones
#define MAX_UNBALANCE_ALLOWED 0.8
#define BALANCING_DELAY 5

extern int size, rank;
extern SortedBuffer * tuples;
extern pthread_mutex_t tuples_mutex;
extern pthread_cond_t idle_cond;
extern bool end;
extern pthread_mutex_t end_flag_mutex;

extern pthread_mutex_t safe_mpi_mutex;
extern MPI_Comm balancingComm;

//double ComputeUnbalanceRate(unsigned long t, int n, int * CS){
double ComputeUnbalanceRate(int t, int n, int * CS){
	//CS: CurrentState
	//BFS: Best Factible Solution
	//WFS: WorstFactibleSolution

	//unsigned long div = t / n;
	int div = (t / n);
	int mod = (t % n);

	//BFS = {ai / ai = (div+1) si 0 <= i <= mod, ai = div si mod <= i < n}
	//WFS = {ai / a0 = t, ai = 0 si 0 <= i < n}


	//Calcular distance(BFS,WFS) ^ 2
	double square_dist_BFS_WFS;
	mod > 0 ? square_dist_BFS_WFS = double((div + 1 - t) ^ 2) + (mod - 1) * ((div + 1) ^ 2) + (n - mod) * (div ^ 2)
			: square_dist_BFS_WFS = double((div - t) ^ 2) + (n - 1) * (div ^ 2);


	//Calcular distance(BFS,CS) ^ 2
	double square_dist_BFS_CS = double(0);

	for (int i = 0; i < mod ; i++){
		square_dist_BFS_CS += double( (div + 1 - CS[i]) ^2);
	}

	for (int i = mod; i < n; i++){
		square_dist_BFS_CS += double((div - CS[i]) ^2);
	}

	return sqrt(square_dist_BFS_CS / square_dist_BFS_WFS);
}



int LoadBalancing(){

	cout<<"Comenzando LoadBalancing, rank = "<<rank<<endl;
	bool exitCondition;
	//unsigned long numPrefs;
	int numPrefs;
	pthread_mutex_lock(&safe_mpi_mutex);
	int currentDestribution[size];
	pthread_mutex_unlock(&safe_mpi_mutex);
	double unbalanceRate;



	do{

		pthread_mutex_lock(&tuples_mutex);
		numPrefs = tuples->size();
		pthread_mutex_unlock(&tuples_mutex);
		cout<<"LoadBalancing: numPrefs: "<<numPrefs<<", size: "<<size<<", rank:"<<rank<<endl;
		//Se distribuyen la cantidad de prefijos largos entre los procesos
		cout<<"LoadBalancing: BEFORE Allgather 1, rank: "<<rank<<endl;
		//pthread_mutex_lock(&safe_mpi_mutex);
		int error = MPI_Allgather(&numPrefs,1,MPI_INT,currentDestribution,1,MPI_INT,balancingComm);
		//pthread_mutex_unlock(&safe_mpi_mutex);
		cout<<"LoadBalancing: AFTER Allgather 1 error: "<<error<<", rank: "<<rank<<endl;
		//Calculo cantidad de prefijos largos en los procesos
		numPrefs = 0;
		for (int i = 0; i < size; i++){
			numPrefs += currentDestribution[i];
		}
		cout<<"LoadBalancing 90: despues de recorrer currentDestribution, numPrefs: "<<numPrefs<<" rank: "<<rank<<endl;

		exitCondition = numPrefs == 0;

		if (exitCondition){
			cout<<"LoadBalancing 95: exitCondition == TRUE "<<" rank: "<<rank<<endl;
			pthread_mutex_lock(&end_flag_mutex);
			end = true;
			pthread_mutex_unlock(&end_flag_mutex);
			cout<<"LoadBalancing 95: end = TRUE "<<" rank: "<<rank<<endl;

		}else{

			//Calcular el indíce de desbalanceo
			cout<<"LoadBalancing 103: BEFORE Calcular el indice de balanceo, rank: "<<rank<<endl;
			unbalanceRate = ComputeUnbalanceRate(numPrefs,size,currentDestribution);
			cout<<"LoadBalancing 105: AFTER Calcular el indice de balanceo: "<<unbalanceRate<<", rank: "<<rank<<endl;
			if (unbalanceRate > MAX_UNBALANCE_ALLOWED) {

				//Calculando parametros necesarios para la distribucion

				int div = numPrefs / size;
				int mod = numPrefs % size;

				int prefsToSend = 0;
				int totalPrefsToDestribute = 0;
				int numSenders = 0;
				int offsetCounter = 0;
				int myOffset = 0; //TODO fijarte por el warning
				int inc;

				for (int i = 0; i < size; i++){

					if (currentDestribution[i] >= div + 1 ){

						numSenders++;
						numSenders > mod ? inc = currentDestribution[i] - div : inc = currentDestribution[i] - div - 1;
						totalPrefsToDestribute += inc;
						if (i == rank){
							prefsToSend = inc;
						}

					}else{

						offsetCounter += div - currentDestribution[i];
						if (i == rank){
							myOffset = offsetCounter;
						}

					}

				}

//				int (*sendBuff)[Tuple::tuplesIntSize];
//				int (*recvBuff)[Tuple::tuplesIntSize];
//				sendBuff = prefsToSend == 0 ? NULL
//						                    : (int (*)[Tuple::tuplesIntSize]) malloc(sizeof(int) * Tuple::tuplesIntSize * prefsToSend);
//				recvBuff = (int (*)[Tuple::tuplesIntSize]) malloc(sizeof(int) * Tuple::tuplesIntSize * totalPrefsToDestribute);
				int sendBuff[prefsToSend][Tuple::tuplesIntSize];
				//int recvBuff[totalPrefsToDestribute][Tuple::tuplesIntSize];
				//int** recvBuff = alloc_2d_int(Tuple::tuplesIntSize, totalPrefsToDestribute);
				int recvBuff[totalPrefsToDestribute][Tuple::tuplesIntSize];
				int recvBuffAux[totalPrefsToDestribute * Tuple::tuplesIntSize];

				cout<<"LoadBalancing: lock tuples_mutex, rank: "<<rank<<endl;
				pthread_mutex_lock(&tuples_mutex);

				if(prefsToSend != 0){
					Tuple * tArray[prefsToSend];
					tuples->popKTuples(prefsToSend,tArray);

					for(int i = 0; i < prefsToSend; i++){
						memcpy(sendBuff[i],tArray[i]->getArray(),Tuple::tuplesIntSize);
						delete tArray[i];
					}
				}


				//Se distribuyen los prefijos entre los procesos
				//pthread_mutex_lock(&safe_mpi_mutex);
				MPI_Allgather(sendBuff,prefsToSend * Tuple::tuplesIntSize,MPI_INT,
						                    recvBuffAux, totalPrefsToDestribute * Tuple::tuplesIntSize, MPI_INT,balancingComm);
				//pthread_mutex_unlock(&safe_mpi_mutex);

				int j = -1;
				for(int i = 0 ; i < (Tuple::tuplesIntSize * size) ; i++){
					if( i % Tuple::tuplesIntSize == 0 ){ j++; }
					recvBuff[j][i % Tuple::tuplesIntSize] = recvBuffAux[i];
				}

				//Si estoy recibiendo prefijos los copio en prefixes desde recvBuff
				int recvQty;
				if ((recvQty = div - currentDestribution[rank])){
					Tuple * tArray[recvQty];

					for (int i = 0; i < recvQty; i++){
						tArray[i] = new Tuple(Tuple::tuplesIntSize,
								                recvBuff[myOffset + i]);
					}

					tuples->insert(tArray,recvQty);
				}

				//Si tengo prefijos nuevos realizo signal sobre la variable de condicion idle_cond por si el thread de procesamiento
				//de tuplas se encuentra detenido a la espera.
				if (tuples->size() > 0){
					cout<<"LoadBalancing: SIGNAL , rank: "<<rank<<endl;
					pthread_cond_signal(&idle_cond);
				}

				pthread_mutex_unlock(&tuples_mutex);

//				free(sendBuff);
//				free(recvBuff);

			}

		}

		//Espero para evaluar el próximo balanceo
		sleep(BALANCING_DELAY);

	}while(!exitCondition);

	//pthread_cond_signal(&idle_cond);
	cout<<"LoadBalancing: END, rank: "<<rank<<endl;
	return 0;
}



